package com.dsk.connector.config;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.io.resource.ResourceUtil;
import cn.sanenen.sunutils.queue.SMQ;
import com.dsk.connector.commons.conf.TableMapping;
import com.dsk.connector.commons.conf.TableRegMapping;
import com.dsk.connector.commons.utils.ConfigUtils;
import com.dsk.connector.sink.common.listener.DatatransMqUtils;
import com.dsk.connector.sink.common.listener.MessageListener;
import com.dsk.connector.source.common.DebeziumHandler;
import io.debezium.embedded.Connect;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.RecordChangeEvent;
import io.debezium.engine.format.ChangeEventFormat;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.connect.source.SourceRecord;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.config.YamlPropertiesFactoryBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;
import org.springframework.core.io.Resource;

import java.io.BufferedReader;
import java.io.File;
import java.lang.reflect.Field;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

@Configuration
public class DebeziumConfig {
    @Bean("tableMappings")
    public Map<String, TableMapping> retrieveTableMapping(Environment env) throws Exception {
        Map<String, TableMapping> tableMapping = new HashMap<>();
        String tableMappingConfig=env.getProperty("config.tableMappingConfig");
        if(StringUtils.isBlank(tableMappingConfig)){
            throw new RuntimeException("未配置tableMappingConfig，请配置config.tableMappingConfig");
        }
        //
        Pattern regTableMappingPattern=Pattern.compile("[a-z0-9A-Z_]+\\.\".+\"");
        Class<TableMapping> TableMappingClass=TableMapping.class;
        BufferedReader reader=ResourceUtil.getUtf8Reader(tableMappingConfig);
        String lineConf=reader.readLine();
        //第一行抽取标题
        String[] fieldNames=lineConf.split(",");
        //从第二行开始读取配置数据
        while((lineConf=reader.readLine())!=null){
            String[] fieldValues=lineConf.split(",");
            TableMapping tableMappingItem=new TableMapping();
            for(int i=0;i<fieldNames.length;i++){
                Field field=TableMappingClass.getDeclaredField(fieldNames[i].trim());
                field.setAccessible(true);
                String originValue= i>=fieldValues.length?"":fieldValues[i];
                Object fieldValue=Convert.convert(field.getType(),originValue);
                field.set(tableMappingItem,fieldValue);
            }
            if(regTableMappingPattern.matcher(tableMappingItem.getSourceTable()).matches()){
                continue;
            }
            if(tableMappingItem.getSourceTable().contains("\"")){
                throw new RuntimeException("sourceTable配置错误，不能包含双引号，或者双引号包括全部正则表达式表名:"+lineConf);
            }
            tableMapping.put(tableMappingItem.getSourceTable(),tableMappingItem);
        }
        return tableMapping;
    }

    @Bean("tableRegMappings")
    public Map<String, TableRegMapping> retrieveTableRegMapping(Environment env) throws Exception {
        Map<String, TableRegMapping> tableMapping = new HashMap<>();
        String tableMappingConfig=env.getProperty("config.tableMappingConfig");
        if(StringUtils.isBlank(tableMappingConfig)){
            throw new RuntimeException("未配置tableMappingConfig，请配置config.tableMappingConfig");
        }
        String enableRegMapping=env.getProperty("config.source.regTableName","false");
        if("false".equalsIgnoreCase(enableRegMapping)){
            return tableMapping;
        }
        //
        Pattern regTableMappingPattern=Pattern.compile("([a-z0-9A-Z_]+)\\.\"(.+)\"");
        Class<TableMapping> TableMappingClass=TableMapping.class;
        BufferedReader reader=ResourceUtil.getUtf8Reader(tableMappingConfig);
        String lineConf=reader.readLine();
        //第一行抽取标题
        String[] fieldNames=lineConf.split(",");
        //从第二行开始读取配置数据
        while((lineConf=reader.readLine())!=null){
            String[] fieldValues=lineConf.split(",");
            TableRegMapping tableMappingItem=new TableRegMapping();
            for(int i=0;i<fieldNames.length;i++){
                Field field=TableMappingClass.getDeclaredField(fieldNames[i].trim());
                field.setAccessible(true);
                String originValue= i>=fieldValues.length?"":fieldValues[i];
                Object fieldValue=Convert.convert(field.getType(),originValue);
                field.set(tableMappingItem,fieldValue);
            }
            //设置源表正则表达式
            String sourceTableName=tableMappingItem.getSourceTable();
            Matcher tableRegMatcher=regTableMappingPattern.matcher(sourceTableName);
            if(!tableRegMatcher.matches()){
                if(sourceTableName.contains("\"")){
                    throw new RuntimeException("sourceTable配置错误，不能包含双引号，或者双引号包括全部正则表达式表名:"+lineConf);
                }
                continue;
            }
            tableMappingItem.setSourceTable(null);
            tableMappingItem.setSourceDb(tableRegMatcher.group(1));
            tableMappingItem.setRegExp(tableRegMatcher.group(2));
            //设置目标表表达式
            Matcher targetTableMatcher=regTableMappingPattern.matcher(tableMappingItem.getTargetTable());
            if(!targetTableMatcher.matches()){
                throw new RuntimeException("正则匹配的目标表配置不合法，请检查配置文件"+tableMappingConfig+"中"+lineConf+"的配置");
            }
            tableMappingItem.setTargetDb(targetTableMatcher.group(1));
            tableMappingItem.setTargetExpression(targetTableMatcher.group(2));
            tableMappingItem.setTargetTable(null);
            //
            tableMapping.put(sourceTableName,tableMappingItem);
        }
        return tableMapping;
    }

    @Bean("debeziumProperties")
    public Properties configureEngine(@Qualifier("tableMappings") Map<String, TableMapping> tableMappings,
                                      @Qualifier("tableRegMappings") Map<String, TableRegMapping> tableRegMappings,
                                      ConfigurableApplicationContext context) {
        Environment environment=context.getEnvironment();
        String debeziumConfig=environment.getProperty("config.debeziumConfig");
        if(StringUtils.isBlank(debeziumConfig)){
            throw new RuntimeException("未配置debeziumConfig，请配置config.debeziumConfig");
        }
        String enableRegMapping=environment.getProperty("config.source.regTableName","false");

        Resource resource=context.getResource(debeziumConfig);

        YamlPropertiesFactoryBean propertiesFactoryBean=new YamlPropertiesFactoryBean();
        propertiesFactoryBean.setResources(resource);
        propertiesFactoryBean.afterPropertiesSet();
        Properties props=propertiesFactoryBean.getObject();
        //占位符处理
        ConfigUtils.setEnvironment(environment);
        ConfigUtils.replacePlaceholders(props);
        //处理要抽取的表和数据库
        Set<String> includeDatabases=new HashSet<>();
        Set<String> includeTables=new HashSet<>();
        for(String tableName:tableMappings.keySet()){
            String[] tableNameArray=tableName.split("\\.");
            if(tableNameArray.length>1){
                includeTables.add(tableName);
            }
            includeDatabases.add(tableNameArray[0]);
        }
        if("true".equalsIgnoreCase(enableRegMapping)){
            for(TableRegMapping tableRegMapping:tableRegMappings.values()){
                includeDatabases.add(tableRegMapping.getSourceDb());
            }
        }

        String includeDatabasesStr=String.join(",",includeDatabases);
        String includeTablesStr=String.join(",",includeTables);
        props.setProperty("database.include.list",includeDatabasesStr);
        if("true".equalsIgnoreCase(enableRegMapping)){
            if(tableRegMappings.size()==0){
                props.setProperty("table.include.list",includeTablesStr);
            }
        }else{
            props.setProperty("table.include.list",includeTablesStr);
        }

        //设置gtid同步
        String gtidEnabled=environment.getProperty("debezium.database.gtid.enabled");
        if("true".equals(gtidEnabled) || "1".equals(gtidEnabled)){
            props.setProperty("read.only","true");
        }
        if("false".equals(gtidEnabled) || "0".equals(gtidEnabled)){
            props.setProperty("read.only","false");
        }
        String offsetFileStr=(String)props.get("offset.storage.file.filename");
        File offsetFile=this.createDirIfNotExist(offsetFileStr,true,"未配置debezium.file.offsetPostion，请在application.yml配置debezium.file.offsetPostion");
        System.out.println("offset.storage.file.filename:"+offsetFile.getAbsolutePath());
        //
        String schemaHistory=(String)props.get("schema.history.internal.file.filename");
        File schemaHistoryFile=this.createDirIfNotExist(schemaHistory,true,"未配置debezium.file.schemaHistory，请在application.yml配置debezium.file.schemaHistory");
        System.out.println("schema.history.internal.file.filename:"+schemaHistoryFile.getAbsolutePath());

        return props;
    }
    @ConditionalOnProperty(name = "config.source.enabled", havingValue = "true")
    @Bean
    public DebeziumEngine configureEngine(@Qualifier("debeziumProperties") Properties props,
                                          @Qualifier("mysqlRecordHandler") DebeziumHandler consumer,
                                          @Qualifier("tableMappings") Map<String, TableMapping> tableMappings,
                                          @Qualifier("tableRegMappings") Map<String, TableRegMapping> tableRegMappings,
                                          CloseApplication closeApplication,
                                          Environment env)  {
        //
        consumer.setTableRegMappings(tableRegMappings);
        consumer.setTableMappings(tableMappings);
        consumer.setEnvironment(env);
        // 使用该配置创建引擎
        DebeziumEngine<RecordChangeEvent<SourceRecord>> engine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class))
                .using(props)
                .notifying(consumer)
                .build();

        closeApplication.setDebeziumEngine(engine);

        return engine;
    }
    @ConditionalOnProperty(name = "config.source.enabled", havingValue = "true")
    @Bean("debeziumExecutor")
    public ExecutorService configExecutorService(DebeziumEngine engine,Environment env,CloseApplication closeApplication){
        //配置mq的缓存目录
        String mqCacheDir=env.getProperty("debezium.file.mqCacheDir");
        File file=this.createDirIfNotExist(mqCacheDir,false,"未配置debezium.file.mqCacheDir，请在application.yml配置debezium.file.mqCacheDir");
        System.out.println("debezium.file.mqCacheDir:"+file.getAbsolutePath());
        SMQ.setting(mqCacheDir);
        //
        ExecutorService executor = Executors.newSingleThreadExecutor();
        executor.execute(engine);

        closeApplication.setDebeziumExecutor(executor);

        return executor;
    }
    protected File createDirIfNotExist(String dir,boolean isFile,String errorMsg){
        if(StringUtils.isBlank(dir)){
            throw new RuntimeException(errorMsg);
        }else{
            File file=new File(dir);
            if(isFile){
                file=file.getParentFile();
            }
            if(!file.exists()){
                file.mkdirs();
            }
            return file;
        }
    }
    @Bean("sinkExecutor")
    public ExecutorService configSinkExecutorService(@Qualifier("tableMappings") Map<String, TableMapping> tableMappings,DatatransMqUtils mqUtils, Environment environment, CloseApplication closeApplication){

        //配置mq的缓存目录
        String mqCacheDir=environment.getProperty("debezium.file.mqCacheDir");
        this.createDirIfNotExist(mqCacheDir,false,"未配置debezium.file.mqCacheDir，请在application.yml配置debezium.file.mqCacheDir");
        SMQ.setting(mqCacheDir);

        if(mqUtils.getTableMappings()==null){
            mqUtils.setTableMappings(tableMappings);
        }
        List<MessageListener> listeners=mqUtils.createMqListener();

        ExecutorService executor = Executors.newFixedThreadPool(listeners.size());
        for(MessageListener listener:listeners){
            executor.execute(listener);
        }
        //
        closeApplication.setSinkExecutor(executor);

        return executor;
    }
}
